package com.yj.rocketmq.producer;


import com.yj.core.exception.ServiceException;
import com.yj.rocketmq.RocketMqConstants;
import com.yj.rocketmq.RocketMqExceptionMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Optional;

/**
 * rocketMq 生成者
 *
 * @author Xie
 * @date 19-3-19
 */
@Slf4j
public abstract class AbstractMqProducer {

    /**
     * 生成对象
     */
    private DefaultMQProducer producer = null;

    /**
     * 返回生产者对象
     *
     * @return
     */
    public DefaultMQProducer getProducer() {
        if (producer == null) {
            log.error("{}  DefaultMQProducer handler is null,不能发送消息", RocketMqConstants.LOG_PREFIX);
            throw new ServiceException(RocketMqExceptionMsg.PRODUCER_HANDLE_NULL);
        }
        return producer;
    }

    /**
     * 初始化
     */
    @PostConstruct
    public void initMQProducer() {
        //检查是否启用
        if (getProducerEnable() == null || !getProducerEnable()) {
            log.warn("{}未启用,无需初始化", RocketMqConstants.LOG_PREFIX);
            return;
        }

        //检查配置参数
        if (StringUtils.isAnyBlank(getNameSrv(), getProducerId())) {
            log.error("{} 请正确配置 DefaultMQProducer参数", RocketMqConstants.LOG_PREFIX);
            throw new ServiceException(RocketMqExceptionMsg.PRODUCER_PARAM_CFG_INVALID);
        }

        //消息发送对象
        producer = new DefaultMQProducer(getProducerId());
        producer.setNamesrvAddr(getNameSrv());
        producer.setVipChannelEnabled(false);
        producer.setRetryTimesWhenSendFailed(Integer.valueOf(getMaxReproducerTimes()).intValue());
        producer.setInstanceName(getProducerId());
        try {
            producer.start();
            log.warn("{}producerId={}开启成功!", RocketMqConstants.LOG_PREFIX, getProducer());
        } catch (MQClientException e) {
            e.printStackTrace();
            log.error("{}启动发生异常，异常信息={}", RocketMqConstants.LOG_PREFIX, e);
            throw new ServiceException(RocketMqExceptionMsg.PRODUCER_START_FAILED);
        }
    }


    /**
     * 在应用退出前, 销毁 Producer 对象
     */
    @PreDestroy
    public void shutDownProducer() {
        if (getProducerEnable() == null
                || !getProducerEnable()
                || producer == null) {
            log.warn("{} DefaultMQProducer未开启，无需关闭", RocketMqConstants.LOG_PREFIX);
            return;
        }
        Optional.ofNullable(producer).ifPresent(defaultMQProducer -> {
            log.info("{}在应用退出前，销毁 DefaultMQProducer　对象", RocketMqConstants.LOG_PREFIX);
            defaultMQProducer.shutdown();
        });
    }


    /**
     * Producer　Id
     *
     * @return
     */
    protected abstract String getProducerId();

    /**
     * mq 地址
     *
     * @return
     */
    protected abstract String getNameSrv();

    /**
     * 重试次数
     *
     * @return
     */
    protected abstract String getMaxReproducerTimes();

    /**
     * 启用标志
     *
     * @return
     */
    protected abstract Boolean getProducerEnable();
}
